#include <brpc/server.h>
#include <butil/logging.h>
#include <thread>

#include "data/mysql.hpp"
#include "data/mysql_chat_session_member.hpp"

#include "channel.hpp"
#include "etcd.hpp"     // 服务注册模块封装
#include "logger.hpp"   // 日志模块封装
#include "rabbitmq.hpp" // 消息队列封装
#include "utils.hpp"

#include "base.pb.h"
#include "user.pb.h"
#include "transmite.pb.h"  // protobuf框架代码

namespace im{
class MsgTransmitServiceImpl :public MsgTransmitService{

public:
    MsgTransmitServiceImpl(const std::string &user_service_name,
            const ServiceManager::ptr &channels,
            const std::shared_ptr<odb::core::database> &mysql_client,
            const std::string &exchange_name,
            const std::string &routing_key,
            const MQClient::ptr &mq_client)
        :_user_service_name(user_service_name),
        _mm_channels(channels),
        _mysql_session_member_table(std::make_shared<ChatSessionMemberTable>(mysql_client)),
        _exchange_name(exchange_name),
        _routing_key(routing_key),
        _mq_client(mq_client){}
     
    virtual void GetTransmitTarget(google::protobuf::RpcController* controller, 
                                    const NewMessageReq* req, 
                                    GetTransmitTargetRsp* rsp, 
                                    google::protobuf::Closure* done){
        LOG_INFO("收到来自 {} 消息转发请求", req->request_id());
        brpc::ClosureGuard rpc_guard(done);
        auto err_response = [this, rsp](const std::string &rid, const std::string &errmsg) -> void {
                rsp->set_request_id(rid);
                rsp->set_success(false);
                rsp->set_errmsg(errmsg);
            };
        std::string rid = req->request_id();
        std::string user_id = req->user_id();
        std::string csid = req->chat_session_id();
        auto msgContent = req->message();
        // 设置响应信息
        // 1.从请求中获取关键信息：用户ID，所属会话ID，消息内容
        rsp->set_request_id(rid);
        rsp->set_success(true);
        rsp->set_errmsg("");
        // 2.给用户子服务发送请求，获取到 发送者的信息
        auto channel = _mm_channels->choose(_user_service_name);
        if(channel == nullptr){
            LOG_CRITICAL("{}-{} 没有可供访问的用户子服务节点！", rid, _user_service_name);
            return err_response(rid, "没有可供访问的用户子服务节点！");
        }
        brpc::Controller cntl;
        GetUserInfoReq request;
        request.set_request_id(rid);
        request.set_user_id(user_id);
        GetUserInfoRsp response;
        UserService_Stub stub(channel.get());
        stub.GetUserInfo(&cntl, &request, &response, nullptr);
        if (cntl.Failed() == true || response.success() == false) {
            LOG_CRITICAL("{} - 用户子服务调用失败：{}！", req->request_id(), cntl.ErrorText());
            return err_response(req->request_id(), "用户子服务调用失败!");
        }
        MessageInfo msg;
        msg.set_message_id(uuid());
        msg.set_chat_session_id(csid);
        msg.set_timestamp(time(nullptr));
        msg.mutable_sender()->CopyFrom(response.user_info());
        msg.mutable_message()->CopyFrom(msgContent);
        rsp->mutable_message()->CopyFrom(msg);

        // 3.通过会话ID可以从数据库里获取所有用户ID（包括发送者）
        auto uid_list = _mysql_session_member_table->members(csid);
        for(auto& uid : uid_list){
            rsp->add_target_id_list(uid);
        }
        
        // 4.消息发到mq消息队列里，让消息存储子服务进行消费，把数据进行持久化
        bool ret = _mq_client->publish(_exchange_name, msg.SerializeAsString(), _routing_key);
        if (ret == false) {
            LOG_CRITICAL("{} - 持久化消息发布失败！", req->request_id());
            return err_response(req->request_id(), "持久化消息发布失败：!");
        }

    }

    
private:
    // 用户子服务相关信息
    std::string _user_service_name;
    ServiceManager::ptr _mm_channels;

    // 聊天会话成员表相关句柄
    ChatSessionMemberTable::ptr _mysql_session_member_table;

    // 消息队列相关信息
    std::string _exchange_name;
    std::string _routing_key;
    MQClient::ptr _mq_client;
};


class MsgTransmitServer{

public:
    using ptr = std::shared_ptr<MsgTransmitServer>;
    MsgTransmitServer(const Discover::ptr discover_client,
                const Registry::ptr &reg_client, 
                const std::shared_ptr<odb::core::database> &client,
                const std::shared_ptr<brpc::Server> &server)
                :_discover_client(discover_client),
                _register_client(reg_client),
                _mysql_client(client),
                _rpc_server(server){}
    ~MsgTransmitServer(){}
    void start(){
        _rpc_server->RunUntilAskedToQuit();
    }

private:
    Discover::ptr _discover_client;     // 服务发现监控端
    Registry::ptr _register_client;     // 服务注册
    std::shared_ptr<odb::core::database> _mysql_client; // mysql客户端
    std::shared_ptr<brpc::Server> _rpc_server;          // rpc服务端
};

class MsgTransmitServerBuild{

public:
    void init_discover_client(const std::string &reg_host,
            const std::string &base_service_name,
            const std::string &user_service_name) {
            _user_service_name = user_service_name;
        _mm_channels = std::make_shared<ServiceManager>();
        _mm_channels->declared(user_service_name);
        LOG_DEBUG("设置用户子服务为需添加管理的子服务：{}", user_service_name);
        auto put_cb = std::bind(&ServiceManager::onServiceOnline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
        auto del_cb = std::bind(&ServiceManager::onServiceOffline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
        _discover_client = std::make_shared<Discover>(reg_host, base_service_name, put_cb, del_cb);
    }
    void init_register_client(const std::string& reg_host, const std::string &service_name, const std::string &access_host){
        _register_client = std::make_shared<Registry>(reg_host);
        _register_client->register_server(service_name, access_host);
    }
    void init_mysql_client(
            const std::string &user, const std::string &pwd,
            const std::string &host, const std::string &db,
            const std::string &cset, int port, int conn_pool_count){
            _mysql_client = ODBFactory::create(user, pwd, host, db, cset, port, conn_pool_count);
        }
    //用于构造rabbitmq客户端对象
    void init_MQ_client(const std::string &user, 
        const std::string &passwd,
        const std::string &host,
        const std::string &exchange_name,
        const std::string &queue_name,
        const std::string &binding_key) {
        _routing_key = binding_key;
        _exchange_name = exchange_name;
        _mq_client = std::make_shared<MQClient>(user, passwd, host);
        _mq_client->declare_component(exchange_name, queue_name, binding_key);
    }

    void init_rpc_server(uint16_t port, int32_t timeout, uint8_t num_threads){
        if (!_mq_client) {
            LOG_CRITICAL("还未初始化 消息队列 模块！");
            abort();
        }
        _rpc_server = std::make_shared<brpc::Server>();
        MsgTransmitServiceImpl *transmite_service = new MsgTransmitServiceImpl(
            _user_service_name, _mm_channels, _mysql_client, _exchange_name, _routing_key, _mq_client);
        int ret = _rpc_server->AddService(transmite_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
        if (ret == -1) {
            LOG_CRITICAL("添加Rpc服务失败！");
            abort();
        }

        brpc::ServerOptions opt;
        opt.num_threads = num_threads;
        opt.idle_timeout_sec = timeout;
        ret = _rpc_server->Start(port, &opt);
        if(ret != 0){
            LOG_CRITICAL("启动服务器失败！");
            abort();
        }
    }

    MsgTransmitServer::ptr build(){
        if (!_mysql_client) {
            LOG_CRITICAL("还未初始化 数据库 模块！");
            abort();
        }
        if (!_discover_client) {
            LOG_CRITICAL("还未初始化 服务发现 模块！");
            abort();
        }
        if (!_register_client) {
            LOG_CRITICAL("还未初始化 服务注册 模块！");
            abort();
        }
        if (!_rpc_server) {
            LOG_CRITICAL("还未初始化 RPC服务器 模块! ");
            abort();
        }
        return std::make_shared<MsgTransmitServer>(_discover_client, _register_client, _mysql_client, _rpc_server);
    }

private:
    std::string _user_service_name;
    ServiceManager::ptr _mm_channels;
    Discover::ptr _discover_client;     // 服务发现监控端
    Registry::ptr _register_client;     // 服务注册
    std::shared_ptr<odb::core::database> _mysql_client; // mysql客户端

    std::string _routing_key;
    std::string _exchange_name;     // 交换机名称
    MQClient::ptr _mq_client;       // rabbitmq客户端

    std::shared_ptr<brpc::Server> _rpc_server;          // rpc服务端
};
}